
package com.trydofor.id.lazy;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;

import javax.annotation.Nonnull;
import javax.annotation.concurrent.ThreadSafe;

import com.trydofor.id.a9cq.ExtedLongA9cq;
import com.trydofor.id.core.A9CQ;
import com.trydofor.id.core.A9Loader;
import com.trydofor.id.core.ApiExceptions;
import com.trydofor.id.core.Range;

/**
 * @author : Shi Rongjiu (www.trydofor.com)
 */
@ThreadSafe
public class LazyLongA9cq implements A9CQ<Long> {
    
    private final A9Loader<Range>  loader;
    private final ThreadFactory    factory;
    private final int              latch;
    private final int              increment;
    
    private final ExtedLongA9cq    elongA9cq  = new ExtedLongA9cq(false, 0);
    private volatile int           procLatch;
    private volatile Future<Range> refFuture;
    private volatile boolean       continuous = true;
    
    /**
     * an atomic lazy long a9cq with default ThreadFactory.
     * 
     * @param increment the increment.
     * @param loader @Nonnull the loader.
     * @param latch the latch.
     * @see #LazyLongA9cq(int, A9Loader, int, ThreadFactory)
     * @see Executors#defaultThreadFactory()
     */
    public LazyLongA9cq(int increment, @Nonnull A9Loader<Range> loader, int latch) {
        this(increment, loader, latch, Executors.defaultThreadFactory());
    }
    
    /**
     * an atomic lazy long a9cq.
     * 
     * @param increment the increment.
     * @param loader @Nonnull the loader.
     * @param latch the latch.
     * @param factory @Nonnull the thread factory
     * @throws NullPointerException if loader of factory is null
     */
    public LazyLongA9cq(int increment, @Nonnull A9Loader<Range> loader, int latch,
            @Nonnull ThreadFactory factory) {
        ApiExceptions.checkNullPointer(loader);
        ApiExceptions.checkNullPointer(factory);
        
        this.increment = increment;
        this.loader = loader;
        this.factory = factory;
        this.latch = latch < 0 ? 0 : latch;
        this.procLatch = this.latch;
    }
    
    /**
     * {@inheritDoc}
     */
    public boolean isAtomic() {
        return true;
    }
    
    /**
     * {@inheritDoc}
     */
    public long remains() {
        return elongA9cq.remains();
    }
    
    /**
     * {@inheritDoc}
     */
    public void reset() {
        elongA9cq.reset();
    }
    
    /**
     * {@inheritDoc}
     */
    public Long current() {
        return elongA9cq.current();
    }
    
    /**
     * @return the continuous
     */
    public boolean isContinuous() {
        return continuous;
    }
    
    /**
     * @param continuous the continuous to set
     */
    public void setContinuous(boolean continuous) {
        this.continuous = continuous;
    }
    
    /**
     * {@inheritDoc}
     */
    public Long next() {
        Long r = null;
        try {
            r = next(0, null);
        }
        catch (InterruptedException e) {
            ApiExceptions.throwFakeException(e);
        }
        catch (TimeoutException e) {
            ApiExceptions.throwFakeException(e);
        }
        return r;
    }
    
    /**
     * {@inheritDoc}
     */
    public Long next(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException {
        
        final long startedNano = System.nanoTime();
        final Thread curThread = Thread.currentThread();
        final boolean needTime = (timeout > 0 && unit != null);
        
        Long result = elongA9cq.next(timeout, unit);
        while (result == null) { // sync load
            try {
                // blocking
                if (needTime) {
                    asyncLoad().get(timeout, unit);
                }
                else {
                    asyncLoad().get();
                }
            }
            catch (ExecutionException e) {
                ApiExceptions.throwFakeException(e);
            }
            
            if (needTime) {
                ApiExceptions.checkThreadInterrupted(curThread);
                ApiExceptions.checkTimeoutNano(startedNano, timeout, unit);
            }
            
            result = elongA9cq.next(timeout, unit);
        }
        
        // check the cache
        final long remain = elongA9cq.remains();
        if (remain < procLatch) {
            asyncLoad();
        }
        return result;
    }
    
    // -----------------
    private Future<Range> asyncLoad() {
        //check loading
        final Future<Range> current = refFuture;
        if (current != null) return current;
        
        // async load
        ExecutorService executor = Executors.newSingleThreadExecutor(factory);
        final AtomicBoolean await = new AtomicBoolean(true);
        final Future<Range> result = executor.submit(new Callable<Range>() {
            
            public Range call() throws Exception {
                final Range nul = null;
                try {
                    Range r = loader.heavyLoad();
                    boolean lb = r.startWith() < r.endupWith();
                    long st = r.startWith();
                    if (!r.includeStart()) {
                        st += lb ? 1 : -1;
                    }
                    long ed = r.endupWith();
                    if (!r.includeEndup()) {
                        ed += lb ? -1 : 1;
                    }
                    
                    if (continuous) { // adjust startWith
                        Long cq = elongA9cq.current();
                        if (cq != null) {
                            long cl = ((r.startWith() - cq) / increment) * increment + cq;
                            if (r.contains(cl)) {
                                st = cl;
                            }
                            else {
                                for (int i = 1; i < 5; i++) {
                                    boolean pc = r.contains(cl + i * increment);
                                    boolean nc = r.contains(cl - i * increment);
                                    if ((pc || nc) && (!pc || !nc)) {
                                        st = cl + (pc ? i : -i) * increment;
                                        break;
                                    }
                                }
                            }
                        }
                    }
                    
                    elongA9cq.append(st, ed, increment);
                    procLatch = latch;
                }
                catch (Exception e) {
                    if (procLatch > 0) {
                        procLatch = -1;
                        loader.handleException(e);
                    }
                    else {
                        throw e;
                    }
                }
                finally {
                    while (await.get()) {
                        // waiting for 'refFuture = result'
                    }
                    refFuture = null;
                }
                return nul; // only a blocking mark
            }
        });
        
        refFuture = result;
        await.set(false);
        executor.shutdown();
        
        return result;
    }
}
